package cn.me.alphamq.nameserver.store;

import cn.me.alphamq.common.store.StopWatchManager;
import cn.me.alphamq.common.sync.StopWatch;
import cn.me.alphamq.nameserver.cache.Container;
import cn.me.alphamq.nameserver.config.cluster.ClusterProperties;
import cn.me.alphamq.nameserver.enumeration.ReplicationModeEnum;
import cn.me.alphamq.nameserver.enumeration.ReplicationRoleEnum;
import cn.me.alphamq.nameserver.event.model.cluster.ReplicationMsgEvent;

import java.util.UUID;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;

/**
 * 读/写同步复制消息队列的管理者，它的两个 public 方法都会阻塞
 *
 * @author f
 */
public class ReplicationMsgQueueManager {

    /**
     * 用于读/写同步复制消息的队列，最多堆积 5000 条消息到队列中
     */
    private final BlockingQueue<ReplicationMsgEvent> replicationMsgQueue = new ArrayBlockingQueue<>(5000);

    /**
     * 将 移除服务实例的消息 放到队列中，在队列已满时会阻塞
     * <p>
     * 在将消息放到队列时不会阻塞，由于可以依赖节点自行检测服务实例是否实效，所以不需要保证数据的强一致性
     * <p>
     * 如果不需要放，则不会放
     *
     * @param key 删除的服务实例在 map 中的键
     */
    public void putRemovedServiceInstance(String key) {
        ClusterProperties clusterProperties = Container.getClusterProperties();
        ReplicationModeEnum mode = clusterProperties.getReplicationModeEnum();
        if (ReplicationModeEnum.STANDALONE == mode) {
            // 如果是单机模式，则无需与其他节点同步
            return;
        }

        if (ReplicationModeEnum.MASTER_SLAVE == mode) {
            if (ReplicationRoleEnum.MASTER == clusterProperties.getMasterReplicationProperties().getRole()) {
                // 主从复制的主节点需要在服务实例的数据发生变化时，同步数据给从节点
                put(new ReplicationMsgEvent(key));
            }
        } else if (ReplicationModeEnum.CHAIN == mode) {
            if (ReplicationRoleEnum.NODE == clusterProperties.getChainReplicationProperties().getRole()) {
                // 链式复制的非尾节点需要在服务实例的数据发生变化时，同步数据给下一个节点
                put(new ReplicationMsgEvent(key));
            }
        }
    }

    /**
     * 将 服务实例数据变化的消息 放到队列中，在队列已满时会阻塞
     * <p>
     * 在将消息放到队列时会阻塞，等待复制的结束，从而保证数据的强一致性
     * <p>
     * 如果不需要放，则不会放
     *
     * @param serviceInstance 数据变化的服务实例
     * @param stopWatch       停表
     */
    public void putShiftyServiceInstance(ServiceInstance serviceInstance, StopWatch stopWatch) {
        ClusterProperties clusterProperties = Container.getClusterProperties();
        ReplicationModeEnum mode = clusterProperties.getReplicationModeEnum();
        if (ReplicationModeEnum.STANDALONE == mode) {
            // 如果是单机模式，则无需与其他节点同步
            return;
        }

        if (ReplicationModeEnum.MASTER_SLAVE == mode) {
            if (ReplicationRoleEnum.MASTER == clusterProperties.getMasterReplicationProperties().getRole()) {
                // 主从复制的主节点需要在服务实例的数据发生变化时，同步数据给从节点
                sendReplicationMsg(serviceInstance, stopWatch);
            }
        } else if (ReplicationModeEnum.CHAIN == mode) {
            if (ReplicationRoleEnum.NODE == clusterProperties.getChainReplicationProperties().getRole()) {
                // 链式复制的非尾节点需要在服务实例的数据发生变化时，同步数据给下一个节点
                sendReplicationMsg(serviceInstance, stopWatch);
            }
        }
    }

    // 发送复制消息，并且停止运行，等待复制完毕
    private void sendReplicationMsg(ServiceInstance serviceInstance, StopWatch stopWatch) {
        // 构建复制消息并发送
        String replicationMsgId = UUID.randomUUID().toString();
        put(new ReplicationMsgEvent(replicationMsgId, serviceInstance));

        // 先停止运行，等待复制完毕
        StopWatchManager.put(replicationMsgId, stopWatch);
        stopWatch.pause();
    }

    // 将复制消息事件放到队列中，在队列已满时会阻塞
    private void put(ReplicationMsgEvent replicationMsgEvent) {
        try {
            // 如果空间不足放不了，则等待到能放的时候
            replicationMsgQueue.put(replicationMsgEvent);
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
    }

    /**
     * 获取复制消息事件，在队列为空时会阻塞
     *
     * @return 队列的第一个复制消息事件
     * @throws InterruptedException 抛出 BlockingQueue 的 take() 方法抛出的异常
     */
    public ReplicationMsgEvent get() throws InterruptedException {
        return replicationMsgQueue.take();
    }
}
